package com.qianbao.finance.sync.modules.elasticsearch.websocket.server;

import static org.jboss.netty.handler.codec.http.HttpHeaders.isKeepAlive;
import static org.jboss.netty.handler.codec.http.HttpHeaders.setContentLength;
import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.OK;
import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.log4j.Logger;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpHeaders.Names;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponse;
import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import org.jboss.netty.util.CharsetUtil;

/**
 * HTTP请求处理器。高度优化
 * 
 * @author WangYuan
 * @date 2016年7月18日上午11:01:00
 */
public class Handler extends SimpleChannelUpstreamHandler {

    private final static String DATASTYLE = "yyyy-MM-dd HH:mm:ss.SSS";

    private final static Logger logger = Logger.getLogger(Handler.class);

    @Override
    public void exceptionCaught(ChannelHandlerContext channelHandlerContext, ExceptionEvent e)
            throws Exception {
        logger.warn("exceptionCaught: " + channelHandlerContext.hashCode(), e.getCause());
        ShareBean.getInstance().closeChannelHandlerContext(channelHandlerContext);
    }

    private static SimpleDateFormat longDateFormater = new SimpleDateFormat(DATASTYLE);

    private String formatLongDate(Date date) {
        try {
            return longDateFormater.format(date);
        } catch (Exception e) {
            logger.error("", e);
            return null;
        }
    }

    @Override
    public void channelClosed(ChannelHandlerContext channelHandlerContext,
            ChannelStateEvent channelStateEvent) throws Exception {
        if (logger.isDebugEnabled()) {
            logger.debug("channelClosed: " + channelHandlerContext.hashCode());
        }
        ShareBean.getInstance().closeChannelHandlerContext(channelHandlerContext);
    }

    @Override
    public void messageReceived(final ChannelHandlerContext channelHandlerContext,
            final MessageEvent e) throws Exception {
        try {
            Object msg = e.getMessage();
            WebSocketServerHandshaker webSocketServerHandshaker;
            if (msg instanceof HttpRequest) {
                HttpRequest httpRequest = (HttpRequest) msg;
                String uri = httpRequest.getUri();
                if (logger.isDebugEnabled()) {
                    logger.debug("uri: " + uri);
                }
/*                String version = httpRequest.getHeader(Names.SEC_WEBSOCKET_VERSION);
                if (logger.isDebugEnabled()) {
                    logger.debug("version: " + version);
                }
                if (version == null) {
                    logger.warn("version: null");
                }*/
                if (uri.equals("/websocket")) {
                    if (logger.isInfoEnabled()) {
                        logger.info("WebSocket握手请求" + channelHandlerContext.hashCode());
                    }
                    WebSocketServerHandshakerFactory webSocketServerHandshakerFactory =
                            new WebSocketServerHandshakerFactory(
                                    this.getWebSocketLocation(httpRequest), null, false);
                    webSocketServerHandshaker =
                            webSocketServerHandshakerFactory.newHandshaker(httpRequest);
                    if (webSocketServerHandshaker == null) {
                        logger.warn("不支持websocket：" + channelHandlerContext.hashCode());
                        webSocketServerHandshakerFactory
                                .sendUnsupportedWebSocketVersionResponse(channelHandlerContext
                                        .getChannel());
                    } else {
                        logger.info("尝试发送握手请求" + channelHandlerContext.hashCode());
                        webSocketServerHandshaker.handshake(channelHandlerContext.getChannel(),
                                httpRequest);
                        logger.info("成功发送握手请求" + channelHandlerContext.hashCode());
                    }
                    return;
                } else if (uri.equals("/favicon.ico")) {
                    DefaultHttpResponse httpResponse = new DefaultHttpResponse(HTTP_1_1, OK);
                    httpResponse.setHeader(CONTENT_TYPE, "text/html; charset=UTF-8");
                    ChannelBuffer channelBuffer =
                            ChannelBuffers.copiedBuffer("", CharsetUtil.UTF_8);
                    setContentLength(httpResponse, channelBuffer.readableBytes());
                    httpResponse.setContent(channelBuffer);
                    ChannelFuture f = channelHandlerContext.getChannel().write(httpResponse);
                    if (!isKeepAlive(httpRequest) || httpResponse.getStatus().getCode() != 200) {
                        f.addListener(ChannelFutureListener.CLOSE);
                    }
                    return;
                } else if (uri.equals("/stat/info")) {
                    DefaultHttpResponse httpResponse = new DefaultHttpResponse(HTTP_1_1, OK);
                    httpResponse.setHeader(CONTENT_TYPE, "text/html; charset=UTF-8");
                    ChannelBuffer channelBuffer =
                            ChannelBuffers.copiedBuffer(ShareBean.getInstance()
                                    .generateChatRoomInfo(), CharsetUtil.UTF_8);
                    setContentLength(httpResponse, channelBuffer.readableBytes());
                    httpResponse.setContent(channelBuffer);
                    ChannelFuture f = channelHandlerContext.getChannel().write(httpResponse);
                    if (!isKeepAlive(httpRequest) || httpResponse.getStatus().getCode() != 200) {
                        f.addListener(ChannelFutureListener.CLOSE);
                    }
                    return;
                } else {
                    sendHttpResponse(channelHandlerContext, httpRequest, new DefaultHttpResponse(
                            HTTP_1_1, FORBIDDEN));
                }
            } else if (msg instanceof WebSocketFrame) {
                WebSocketFrame webSocketFrame = (WebSocketFrame) msg;
                if (webSocketFrame instanceof CloseWebSocketFrame) {
                    if (logger.isInfoEnabled()) {
                        logger.info("WebSocket连接关闭请求" + channelHandlerContext.hashCode());
                    }
                    ShareBean.getInstance().closeChannelHandlerContext(channelHandlerContext);
                    return;
                } else if (webSocketFrame instanceof PingWebSocketFrame) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("WebSocket Ping请求" + channelHandlerContext.hashCode());
                    }
                    channelHandlerContext.getChannel().write(
                            new PongWebSocketFrame(webSocketFrame.getBinaryData()));
                    return;
                } else if (msg instanceof TextWebSocketFrame) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("WebSocket 加入聊天室请求" + channelHandlerContext.hashCode());
                    }
                    String request = ((TextWebSocketFrame) msg).getText();
                    String[] strs = request.split(" ");
                    if (strs.length < 2) {
                        logger.error("非法的参数: " + request);
                        return;
                    }
                    String cmd = strs[0];
                    if (cmd.equals("join") == false) {
                        logger.error("非法命令: " + cmd + ", 目前只支持join命令");
                        return;
                    }
                    String chatRoomName = strs[1];
                    if (cmd.equals("join") == false) {
                        logger.error("chatRoomName: " + chatRoomName);
                        return;
                    }
                    String groups = ChatRoom.DEFAULT_GROUP;
                    if (strs.length == 3) {
                        groups = strs[2];
                    }
                    logger.info("groups: " + groups);
                    if (cmd.equals("join") == false) {
                        logger.error("chatRoomName: " + chatRoomName);
                        return;
                    }
                    ChatRoom chatRoom = ShareBean.getInstance().getChatRoom().get(chatRoomName);
                    if (chatRoom == null) {
                        logger.info("聊天室: " + chatRoomName + "不存在，需要创建");
                        chatRoom = new ChatRoom();
                        chatRoom.setChatRoomName(chatRoomName);
                        ShareBean.getInstance().getChatRoom().put(chatRoomName, chatRoom);
                    }
                    chatRoom.addChannelHandlerContext(groups, channelHandlerContext);
                    if (logger.isDebugEnabled()) {
                        logger.debug("成功加入聊天室: " + chatRoomName + " "
                                + channelHandlerContext.hashCode());
                    }
                    try {
                        StringBuffer sf = new StringBuffer();
                        sf.append("{");
                        sf.append("\"raw\":\"join_ok\"");
                        sf.append(",\"connectionId\":\"").append(channelHandlerContext.hashCode())
                                .append("\"");
                        sf.append(",\"sysTime\":\"").append(this.formatLongDate(new Date()))
                                .append("\"");
                        sf.append(",\"chatRoomName\":\"").append(chatRoomName).append("\"");
                        sf.append(",\"groups\":\"").append(groups).append("\"");
                        sf.append("}");
                        final TextWebSocketFrame textWebSocketFrame =
                                new TextWebSocketFrame(sf.toString());
                        Channel channel = channelHandlerContext.getChannel();
                        if (channel.isOpen() && channel.isConnected()) {
                            if (logger.isDebugEnabled()) {
                                logger.debug("join_ok发送消息：" + channelHandlerContext.hashCode());
                            }
                            channel.write(textWebSocketFrame);
                        }
                    } catch (Exception ee) {
                        logger.error("", ee);
                    }
                } else {
                    logger.warn("未知协议");
                }
            } else {
                logger.warn("未知协议");
            }
        } catch (Exception ee) {
            logger.error("", ee);
            Object msg = e.getMessage();
            if (msg instanceof HttpRequest) {
                logger.error("uri: " + ((HttpRequest) msg).getUri());
            }
            ShareBean.getInstance().closeChannelHandlerContext(channelHandlerContext);
        }
    }

    private void sendHttpResponse(ChannelHandlerContext channelHandlerContext,
            HttpRequest httpRequest, HttpResponse httpResponse) {
        if (httpResponse.getStatus().getCode() != 200) {
            httpResponse.setContent(ChannelBuffers.copiedBuffer(
                    httpResponse.getStatus().toString(), CharsetUtil.UTF_8));
            setContentLength(httpResponse, httpResponse.getContent().readableBytes());
        }
        ChannelFuture f = channelHandlerContext.getChannel().write(httpResponse);
        if (!isKeepAlive(httpRequest) || httpResponse.getStatus().getCode() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    private String getWebSocketLocation(HttpRequest httpRequest) {
        return "ws://" + httpRequest.getHeader(HttpHeaders.Names.HOST) + "/websocket";
    }
}
